package com.zy.explore.report.application.event.publish;

import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.util.MimeTypeUtils;

import java.util.stream.Collectors;
import java.util.stream.Stream;

@Component
public class MqSender {

    @Autowired
    private MessageOutput source;

    /**
     * 1发送普通字符串消息
     *@param msg
     *@throws Exception
     */
    public boolean send(String msg) {
        return source.output1().send(MessageBuilder.withPayload(msg).build());
    }

    /**
     * 2发送带tag的消息
     *@param msg
     *@param tag
     *@throws Exception
     */
    public boolean sendWithTags(String msg, String tag) {
        Message message = MessageBuilder.createMessage(msg, new MessageHeaders(Stream.of(tag)
                .collect(Collectors.toMap(str -> MessageConst.PROPERTY_TAGS, String::toString))));
        return source.output1().send(message);
    }

    /**
     * 3发送对象消息
     *@param msg
     *@param tag
     *@param <T>
     */
    public <T>T sendObject(T msg, String tag) {
        Message message = MessageBuilder.withPayload(msg)
                .setHeader(MessageConst.PROPERTY_TAGS, tag)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build();
        boolean result = source.output1().send(message);
        if (result) {
            return msg;
        }
        return null;
    }
    /**
     * 4发送事务消息
     *@param msg
     *@param num
     *@param <T>
     *@throws Exception
     */
    public <T> T sendTransactionalMsg(T msg, int num) {
        MessageBuilder builder = MessageBuilder.withPayload(msg)
                .setHeader(MessageHeaders.CONTENT_TYPE,
                        MimeTypeUtils.APPLICATION_JSON);
        builder.setHeader("type", String.valueOf(num));
        Message message = builder.build();
        boolean result = source.output2().send(message);
        if (result) {
            return msg;
        }
        return null;
    }

    /**
     * 发送消息到pull的目的地址，为了演示我们消息的手动拉取
     *
     * @param msg 消息体
     */
    public boolean sendPullMessage(String msg) {
        return source.output3().send(MessageBuilder.withPayload(msg).build());
    }

}
